#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import json
import sys
from contextlib import contextmanager
from functools import wraps
from inspect import signature

import pexpect

from common.cleaner import clear
from common.common import check_sql_cmd_param, execute_cmd, check_command_injection
from common.const import SysData, CMDResult
from common.util.check_user_utils import check_os_user
from common.exception.common_exception import ErrCodeException
from common.util.cmd_utils import cmd_format
from common.util.exec_utils import exec_mkdir_cmd, su_exec_cmd_list, ExecFuncParam, exec_ln_cmd
from common.util.validators import ValidatorEnum
from generaldb.gbase8a.common.common_models import ModifyPermission
from generaldb.gbase8a.common.const import RoachConstant, EnvName, ErrorCode, PexpectResult
from generaldb.gbase8a.common.gbase_common import check_path, get_uid_gid
from generaldb.gbase8a.util.gbase_util import GbaseUtil


def register_check(*check_args, **check_kwargs):
    """
    注册校验规则函数
    check_args: tuple[check_func]
    check_kwargs: {check_func_name:check_func}
    :return: wrapper
    e.g:  register_check(check_func1,path_check=path_check_func,cmd_check=cmd_check_func)
    """

    def decorate(func):
        sig = signature(func)
        bound_types = sig.bind_partial(*check_args, **check_kwargs).arguments

        @wraps(func)
        def wrapper(*args, **kwargs):
            bound_val = sig.bind(*args, **kwargs).arguments
            for name, val in bound_val.items():
                if name in bound_types:
                    if val is None:
                        del kwargs[name]
                    else:
                        kwargs[name] = (bound_types[name], val)
            return func(*args, **kwargs)

        return wrapper

    return decorate


class Checker:
    @staticmethod
    def sql_check(sql_arg):
        if not check_sql_cmd_param(sql_arg):
            raise ErrCodeException(err_code=ErrorCode.ERROR_PARAM, message="Invalid sql cmd params")

    @staticmethod
    def path_check(path_arg):
        if not check_path(path_arg):
            raise ErrCodeException(err_code=ErrorCode.ERROR_PARAM, message="Invalid path params")

    @staticmethod
    def cmd_check(cmd_arg):
        if check_command_injection(cmd_arg):
            raise ErrCodeException(err_code=ErrorCode.ERROR_PARAM, message="Invalid cmd params")

    @classmethod
    @register_check(
        path_args=path_check,
        sql_args=sql_check,
        cmd_args=cmd_check
    )
    def check_args(cls, path_args=None, cmd_args=None, sql_args=None):

        def decorate(func):
            sig = signature(func)
            chk_args = [arg for arg in (path_args, cmd_args, sql_args) if arg is not None]
            # 获取待检验参数名与校验函数映射关系
            unpacked_args = {
                arg_name: check_func
                for (check_func, check_args) in chk_args
                for arg_name in check_args
            }

            @wraps(func)
            def wrapper(*args, **kwargs):
                bound_val = sig.bind(*args, **kwargs).arguments
                for name, val in bound_val.items():
                    if name in unpacked_args:
                        check_func = unpacked_args.get(name)
                        if isinstance(check_func, staticmethod):
                            exe_func = check_func.__func__
                        elif isinstance(check_func, classmethod):
                            exe_func = check_func.__func__
                        else:
                            exe_func = check_func
                        if not callable(exe_func):
                            raise ErrCodeException(
                                err_code=ErrorCode.SYSTEM_ERROR,
                                message="Unexpected function to register on check wrapper."
                            )
                        if val:
                            exe_func(val)
                return func(*args, **kwargs)

            return wrapper

        return decorate


class GbaseCmd:
    """
    Gbase 功能实现公共命令
    """

    def __init__(self, pid=None):
        self._base_cmd = "su - {} -c \"{}\""
        self._timeout = 60
        self.pid = pid
        self._init_pwd()

    @staticmethod
    def recursion_modifying_permissions(path_info: ModifyPermission):
        """
        递归修改权限
        :param path_info: 路径信息
        :return: bool, str
        """
        if path_info.user:
            uid, gid = get_uid_gid(path_info.user)
            if not uid and not gid:
                return False, ''
            cmd = "chown {}:{} -R {}".format(uid, gid, path_info.path)
            ret, out = GbaseCmd._execute_cmd(cmd)
            if not ret:
                return False, out
        if path_info.mode:
            cmd = "chmod {} -R {}".format(path_info.mode, path_info.path)
            ret, out = GbaseCmd._execute_cmd(cmd)
            if not ret:
                return False, out
        return True, ''

    @staticmethod
    def get_os_pwd_key():
        return "os_password"

    @staticmethod
    def su_exec_cmd(user, cmd, param_list=None):
        """
        指定用户执行命令通用方法
        :param user: 用户
        :param cmd： 命令
        :param cmd： 校验列表
        :return:
        """
        if param_list is None:
            param_list = [[]]
        exec_param = ExecFuncParam(user, [cmd], param_list, chk_exe_owner=False)
        ret, out = su_exec_cmd_list(exec_param)
        return ret == CMDResult.SUCCESS, out

    @staticmethod
    def _execute_cmd(cmd):
        return_code, out_info, err_info = execute_cmd(cmd)
        ret = (return_code == "0")
        res_cont = out_info if ret else err_info
        return ret, res_cont

    def get_stdin_field(self, field):
        input_dict = json.loads(SysData.SYS_STDIN)
        val = input_dict.get(field)
        del input_dict
        return val

    def get_db_user(self):
        db_user_key = self.get_db_key()
        return self.get_stdin_field(db_user_key)

    def get_os_user(self):
        os_user = self.get_stdin_field("os_username")
        if not os_user:
            input_dict = json.loads(SysData.SYS_STDIN)
            for key in input_dict:
                if "appEnv_auth_authKey" in key:
                    os_user = input_dict.get(key)
                    break
            del input_dict
        if not check_os_user(os_user):
            raise ErrCodeException(err_code=ErrorCode.ERROR_PARAM, message="Invalid os_user")
        return os_user

    def get_custom_key(self):
        return "_".join((EnvName.CUSTOM_SETTINGS, self.pid))

    def get_db_key(self):
        return "_".join((EnvName.DB_USER_NAME, self.pid))

    def get_db_pwd_key(self):
        return "_".join((EnvName.DB_PASSWORD, self.pid))

    def get_db_pwd(self):
        key = self.get_db_pwd_key()
        return self.get_stdin_field(key)

    def get_os_pwd(self):
        key = GbaseCmd.get_os_pwd_key()
        return self.get_stdin_field(key)

    @Checker.check_args(cmd_args=["db_user"])
    def show_gcluster_mode(self, db_user):
        show_mode_cmd = "gcadmin"
        return self.su_exec_cmd(db_user, show_mode_cmd)

    @Checker.check_args(cmd_args=["db_user"])
    def get_db_version(self, db_user):
        version_cmd = "gbase -V"
        return self.su_exec_cmd(db_user, version_cmd)

    @Checker.check_args(cmd_args=["db_user", "env_name"])
    def get_db_os_env(self, db_user, env_name):
        """获取用户的指定环境变量，非root用户无法通过标准库获取"""
        db_env_cmd = "echo ${env_name}"
        return self.su_exec_cmd(db_user, db_env_cmd, [[("env_name", env_name, ValidatorEnum.CHAR_CHK_COMMON)]])

    @Checker.check_args(
        path_args=["src", "symlink_path"],
        cmd_args=["db_user", "src", "symlink_path"])
    def create_datapath_symlink(self, db_user, src, symlink_path):
        return exec_ln_cmd(src, symlink_path, db_user)

    @Checker.check_args(
        path_args=["symlink_path"],
        cmd_args=["db_user", "symlink_path"])
    def delete_datapath_symlink(self, db_user, symlink_path):
        rm_cmd = "rm {symlink_path}"
        return self.su_exec_cmd(db_user, rm_cmd, [[("symlink_path", symlink_path, ValidatorEnum.PATH_CHK_REPO)]])

    @Checker.check_args(
        sql_args=["vc_name"],
        cmd_args=["db_os_user", "db_user", "vc_name"])
    def show_databases(self, db_os_user, db_user, expect_envs: [tuple], vc_name=None):
        """

        :param db_os_user:
        :param db_user:
        :param expect_envs: [(key,[expect_word1,expect_word2]),……]
        :param vc_name:vc name
        :return:
        """
        use_vc = ""
        if vc_name:
            use_vc = cmd_format("use vc {}; ", vc_name)
        show_cmd = f"gbase -u {db_user} -p -e '{use_vc}show databases;'"
        cmd = self._base_cmd.format(db_os_user, show_cmd)
        return self._execute_cmd_with_expect(cmd, expect_envs)

    @Checker.check_args(
        sql_args=["db_name", "vc_name"],
        cmd_args=["db_os_user", "db_user", "db_name", "vc_name"])
    def query_database(self, db_os_user, db_user, db_name, expect_envs, vc_name=None):
        """

        :param db_os_user:
        :param db_user:
        :param db_name:
        :param expect_envs: [(key,[expect_word1,expect_word2]),……]
        :return:
        """
        use_vc = ""
        if vc_name:
            use_vc = cmd_format("use vc {}; ", vc_name)
        drop_cmd = f"gbase -u {db_user} -p -e '{use_vc}use {db_name};'"
        cmd = self._base_cmd.format(db_os_user, drop_cmd)
        return self._execute_cmd_with_expect(cmd, expect_envs)

    @Checker.check_args(
        sql_args=["db_name", "vc_name"],
        cmd_args=["db_os_user", "db_user", "db_name", "vc_name"])
    def drop_database(self, db_os_user, db_user, db_name, expect_envs, vc_name=None):
        """

        :param db_os_user:
        :param db_user:
        :param db_name:
        :param expect_envs: [(key,[expect_word1,expect_word2]),……]
        :return:
        """
        use_vc = ""
        if vc_name:
            use_vc = cmd_format("use vc {}; ", vc_name)
        drop_cmd = f"gccli -u {db_user} -p -e '{use_vc}drop database {db_name};'"
        cmd = self._base_cmd.format(db_os_user, drop_cmd)
        return self._execute_cmd_with_expect(cmd, expect_envs, time_out=False)

    @Checker.check_args(
        sql_args=["vc_name"],
        cmd_args=["db_os_user", "db_user", "vc_name"])
    def refresh_database(self, db_os_user, db_user, expect_envs, vc_name=None):
        """

        :param db_os_user:
        :param db_user:
        :param expect_envs: [(key,[expect_word1,expect_word2]),……]
        :return:
        """
        use_vc = ""
        if vc_name:
            use_vc = cmd_format("use vc {}; ", vc_name)
        refresh_cmd = f"gbase -u {db_user} -p -e '{use_vc}refresh tables;'"
        cmd = self._base_cmd.format(db_os_user, refresh_cmd)
        return self._execute_cmd_with_expect(cmd, expect_envs, time_out=False)

    @Checker.check_args(
        sql_args=["db_name", "vc_name"],
        cmd_args=["gcrcman", "db_os_user", "symlink", "db_name", "vc_name"])
    def restore_database(self, param, env_keys):
        """

        :param env_keys: [(key,[expect_word1,expect_word2]),……]
        :param param 恢复命令参数
        :return:
        """
        db_name = param.db_name
        if param.vc:
            db_name = f"{param.vc}.{param.db_name} "
        drop_cmd = cmd_format("{} -d {} -P -p -e ", param.gcrcman,
                              param.copy_dir) + f"'recover database {db_name} {param.cycle_id} {param.point_id}'"
        cmd = self._base_cmd.format(param.db_os_user, drop_cmd)
        return self._execute_cmd_with_expect(cmd, env_keys, time_out=False)

    @Checker.check_args(
        sql_args=["db_name", "vc_name"],
        cmd_args=["db_os_user", "db_user", "db_name", "vc_name"])
    def get_db_engine_exp(self, db_os_user, db_user, db_name, expect_envs, vc_name=None):
        """
        获取指定数据库为非express的引擎
        :param db_os_user:
        :param db_user:
        :param db_name:
        :param expect_envs: [(key,[expect_word1,expect_word2]),……]
        :param vc_name:
        :return:
        """
        use_vc = ""
        if vc_name:
            use_vc = f"use vc {vc_name}; "
        sql = (f"select distinct engine from information_schema.tables where table_schema='\\''{db_name}'\\'' "
               f"and ( (engine != '\\''EXPRESS'\\'' and  table_type='\\''BASE TABLE'\\'' "
               f"and VC_ID=(select vc()) ) or length(vc_id)=0 )")
        engin_cmd = f"gbase -u {db_user} -p -e '{use_vc}{sql};{chr(92)}'"
        cmd = self._base_cmd.format(db_os_user, engin_cmd)
        return self._execute_cmd_with_expect(cmd, expect_envs, time_out=False)

    @Checker.check_args(
        path_args=["gcrcman"],
        cmd_args=["gcrcman", "db_os_user", "symlink"])
    def show_backup(self, gcrcman, db_os_user, symlink, envs: [tuple]):
        """

        :param gcrcman:
        :param db_os_user:
        :param symlink:
        :param envs: [(key,[expect_word1,expect_word2]),……]
        :return:
        """
        show_backup_cmd = cmd_format("{} -d {} -P -p -e 'show backup'", gcrcman, symlink)
        cmd = self._base_cmd.format(db_os_user, show_backup_cmd)
        return self._execute_cmd_with_expect(cmd, envs, time_out=False)

    @Checker.check_args(
        cmd_args=["db_os_user"]
    )
    def show_cluster_coordinary(self, db_os_user):
        return self.su_exec_cmd(db_os_user, RoachConstant.COORDINATOR_CLUSTER)

    @Checker.check_args(
        cmd_args=["db_os_user"]
    )
    def show_cluster_gcware(self, db_os_user):
        return self.su_exec_cmd(db_os_user, RoachConstant.GCWARE_CLUSTER)

    @Checker.check_args(
        cmd_args=["db_os_user"]
    )
    def show_cluster_data(self, db_os_user):
        return self.su_exec_cmd(db_os_user, RoachConstant.DATA_CLUSTER)

    @Checker.check_args(
        cmd_args=["os_user", "path"],
        path_args=["path"]
    )
    def mk_user_dir(self, path, os_user):
        return exec_mkdir_cmd(path, os_user)

    def check_os_user(self):
        """
        :return:
        """
        envs = [
            (GbaseCmd.get_os_pwd_key(), PexpectResult.OS_LOGIN)
        ]
        user_name = self.get_os_user()
        check_cmd = self._base_cmd.format(user_name, cmd_format("su - {} -c 'exit'", user_name))
        return self._execute_cmd_with_expect(check_cmd, envs)

    @contextmanager
    def _safe_get_key(self, env_key):
        yield env_key
        clear(env_key)

    def _init_pwd(self):
        """ 获取只需用的字段信息，并清除其余字段敏感信息"""
        input_string = sys.stdin.readline()
        input_dict = json.loads(input_string)
        clear(input_string)
        if not input_dict:
            return
        db_user_key = self.get_db_key()
        db_pwd_key = self.get_db_pwd_key()
        custom_key = self.get_custom_key()
        custom_setting = input_dict.get(custom_key)
        custom_dict = {}
        if custom_setting:
            custom_dict.update(GbaseUtil.parse_custom_str_to_dict(custom_setting))
            clear(custom_setting)
        # 清除非必要敏感字段信息
        for field in list(input_dict.keys()):
            if field not in (db_pwd_key, db_user_key):
                del input_dict[field]
        input_dict.update(custom_dict)
        SysData.SYS_STDIN = json.dumps(input_dict)
        del input_dict

    def _execute_cmd_with_expect(self, cmd, env_keys: [tuple], time_out=True):
        """

        :param cmd:
        :param env_keys: 包含密码和预期值的元组列表[(key_field,[expect_world1,expect_world2]),……]
        :return:
        """
        if time_out:
            timeout = self._timeout
        else:
            timeout = None
        child = pexpect.spawn(cmd, encoding='utf-8', timeout=timeout)
        default_expect_worlds = [pexpect.TIMEOUT, pexpect.EOF]
        expect_index = len(default_expect_worlds)
        for env_key, expect_words in env_keys:
            default_expect_worlds.extend(expect_words)
            create_socket_code = child.expect(default_expect_worlds)
            result = child.before
            if create_socket_code < expect_index:
                return False, result
            with self._safe_get_key(self.get_stdin_field(env_key)) as key:
                child.sendline(s=key)
        result = child.read()
        child.close()
        ret = child.exitstatus
        return ret == 0, result
